package com.znyj.servlet.kafka;

import org.apache.commons.lang.SerializationUtils;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

public class KafkaClt {
    Map<String,Object> m_config = null;
    //Producer<String, byte[]> m_producer;

    Producer<String, String> m_producer;
    public KafkaClt(String host){
        boolean r = this.connect(host);
        if (r == false){
            return;
        }
    }

    public boolean connect(String host){
        this.m_config = new HashMap<String, Object>();
        this.m_config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, host);
        m_config.put(ProducerConfig.LINGER_MS_CONFIG,0);
        m_config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000);
        m_config.put(ProducerConfig.RETRIES_CONFIG, "1");
        m_config.put(ProducerConfig.ACKS_CONFIG, "1");
        m_config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        m_config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        try {
            m_producer = new KafkaProducer<String, String>(this.m_config);
        }catch (Exception e){
            System.out.println("connect fail" + e.getMessage());
            return false;
        }

        return false;
    }

    public boolean sendMsg(String deq_name, String msg){
        this.m_producer.send(new ProducerRecord<String, String>(deq_name, 0, "0", msg),
                new Callback() {
                    public void onCompletion(RecordMetadata metadata, Exception e) {
                        if (e != null) {
                            e.printStackTrace();
                            System.out.println("The offset of the record we just sent is: " +
                                    metadata.offset() + "---" + metadata.topic() + "--" + metadata.partition());
                        }
                    }
                });
        return true;
    }

    public void close() {
        m_producer.close();
    }

}